package cn.yanceysong.doris.task;

import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class TaskInfoToTaskReport {
    public static void main(String[] args) throws Exception {
        // 1. 创建 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 2. doris 同步流式写入必须开启 Checkpoint
        // https://doris.apache.org/zh-CN/docs/ecosystem/flink-doris-connector#%E5%B8%B8%E8%A7%81%E9%97%AE%E9%A2%98
        env.enableCheckpointing(1000); // 每 1000 毫秒触发一次 Checkpoint
        env.getCheckpointConfig().setCheckpointingConsistencyMode(CheckpointingMode.EXACTLY_ONCE); // 设置 Exactly-Once 语义
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); // 两次 Checkpoint 之间的最小间隔
        env.getCheckpointConfig().setCheckpointTimeout(60000); // Checkpoint 超时时间
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // 最大并发 Checkpoint 数量
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); // 允许的 Checkpoint 失败次数

        // 3. 创建 MySQL CDC 表
        String mysqlSourceTable = "CREATE TABLE gms_task (\n" +
                                  "    id BIGINT,\n" +
                                  "    type_id INT,\n" +
                                  "    assigned TIMESTAMP,\n" +
                                  "    status STRING,\n" +
                                  "    assignee_id int,\n" +
                                  "    team_id int\n" +
                                  ") WITH (\n" +
                                  "    'connector' = 'mysql-cdc',\n" +
                                  "    'hostname' = 'rm-uf639588047126dlb.mysql.rds.aliyuncs.com',\n" +
                                  "    'port' = '3306',\n" +
                                  "    'username' = 'gms_dev',\n" +
                                  "    'password' = 'aJlhjEb4ErYQ3mjU',\n" +
                                  "    'database-name' = 'gms_dev',\n" +
                                  "    'table-name' = 'gms_task',\n" +
                                  "    'scan.startup.mode' = 'initial',\n" +
                                  "    'scan.incremental.snapshot.chunk.key-column' = 'id',\n" +
                                  "    'scan.incremental.snapshot.enabled' = 'true'\n" +
                                  ")\n";
        tableEnv.executeSql(mysqlSourceTable);

        // 4. 创建 Doris 表（通过 doris 连接） 不要使用jdbc连接，会语法报错。。。
        String dorisSinkDDL = "CREATE TABLE `gms_task_report`  (\n" +
                              "  `task_id` bigint ,\n" +
                              "  `task_type_id` bigint ,\n" +
                              "  `assigned` TIMESTAMP NULL,\n" +
                              "  `status` varchar(255) ,\n" +
                              "  `user_name` varchar(600) ,\n" +
                              "  `user_id` bigint ,\n" +
                              "  `designer_level_id` bigint ,\n" +
                              "  `team_id` bigint ,\n" +
                              "  `level_code` varchar(255) ,\n" +
                              "    PRIMARY KEY (task_id) NOT ENFORCED\n" +
                              ") WITH (\n" +
                              "    'connector' = 'doris',\n" +
                              "    'fenodes' = '172.22.76.131:8030',\n" +
                              "    'table.identifier'= 'mts2_sit.gms_task_report',\n" +
                              "    'username' = 'mts2_sit',\n" +
                              "    'password' = 'FgJW0zcKqCy5qyDP'\n" +
                              ")";
        tableEnv.executeSql(dorisSinkDDL);

        // 5. 查询 MySQL 表并写入 Doris
        String query =
                "INSERT INTO gms_task_report (\n" +
                "    task_id, task_type_id, assigned, status, user_name, user_id, designer_level_id, team_id, level_code\n" +
                ")\n" +
                "SELECT\n" +
                "    task.id AS task_id,\n" +
                "    task.type_id AS task_type_id,\n" +
                "    task.assigned AS assigned,\n" +
                "    task.status AS status,\n" +
                "    report.user_name AS user_name,\n" +
                "    task.assignee_id AS user_id,\n" +
                "    report.designer_level_id AS designer_level_id,\n" +
                "    task.team_id AS team_id,\n" +
                "    report.level_code AS level_code\n" +
                "FROM gms_task AS task \n" +
                "LEFT JOIN gms_task_report AS report ON task.id = report.task_id;\n";

        tableEnv.executeSql(query);
        //        // 3. 将 MySQL 表转换为 DataStream
//        Table mysqlTable = tableEnv.from("mysql_source");
//        DataStream<Row> dataStream = tableEnv.toChangelogStream(mysqlTable);
//
//        // 4. 将 DataStream 写入 Doris
//        dataStream.addSink(new DorisSinkFunction());
//        // 4. 查询 MySQL 表并写入 Doris


        // 6. 启动 Flink 作业
        env.execute("MySQL Binlog to Doris with Table API");
    }
}